package com.chukun.flink.dataset.param;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

/**
 * @author chukun
 * @version 1.0.0
 * @description 参数操作
 * @createTime 2022年05月28日 23:47:00
 */
public class ParamaterTemplate {

    public static void main(String[] args) throws Exception {

        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<Long> dataSource = env.generateSequence(1, 10);

        Configuration cofig = new Configuration();
        cofig.setInteger("limit", 6);

        DataSet<Long> filterDataSet = dataSource
                .filter(new FilterWithParameters())
                .withParameters(cofig);

        filterDataSet.print("withParameters");

        env.execute("ParamaterTemplate");

    }

    private static class FilterWithParameters extends RichFilterFunction<Long> {

        // 从config中获取参数
        private int limit;

        @Override
        public boolean filter(Long value) throws Exception {
            return value > limit;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            this.limit = parameters.getInteger("limit", 2);
        }
    }
}
